package network

import (
	"leaf/core/log"
	"leaf/core/msg"
	"leaf/core/tool"
	"net"
	"sync"
	"sync/atomic"
	"time"
)

type TcpConn struct {
	sync.Mutex
	addr      string
	conn      net.Conn
	writeChan chan []byte
	closeFlag bool
	rflag     int32 //读写线程标识
	cflag     int32 //连接线程标识
	dflag     int32 //销毁标识
	//state     int32
	msgParser *MsgParser
	rwg       sync.WaitGroup
	cwg       sync.WaitGroup
	connType  int32
	server    *TcpServer
	agent     Agent
}

func newTcpAcceptConn(conn net.Conn, msgParser *MsgParser, server *TcpServer) *TcpConn {
	tcpConn := new(TcpConn)
	tcpConn.conn = conn
	if conn == nil {
		log.Error("new accept TcpConn err: conn is nil")
		return nil
	}
	tcpConn.writeChan = make(chan []byte, 1025)
	tcpConn.msgParser = msgParser
	if tcpConn.msgParser == nil {
		tcpConn.msgParser = DefaultMsgParser
	}
	tcpConn.connType = int32(CONN_TYPE_ACCEPT)
	tcpConn.server = server
	return tcpConn
}

func NewTcpConn(addr string, msgParser *MsgParser, agent Agent) *TcpConn {
	tcpConn := new(TcpConn)
	tcpConn.addr = addr
	tcpConn.writeChan = make(chan []byte, 1025)
	tcpConn.msgParser = msgParser
	if tcpConn.msgParser == nil {
		tcpConn.msgParser = DefaultMsgParser
	}
	tcpConn.connType = int32(CONN_TYPE_CONNECT)
	tcpConn.agent = agent
	if tcpConn.agent == nil {
		log.Error("new TcpConn err: agent is nil")
		return nil
	}
	return tcpConn
}

func (tcpConn *TcpConn) dial() net.Conn {
	for {
		if tcpConn.cflag != 1 {
			log.Release("tcpConn dial flag false")
			return nil
		}

		conn, err := net.Dial("tcp", tcpConn.addr)
		if err == nil {
			if tcpConn.cflag != 1 {
				log.Release("tcpConn dial flag false")
				conn.Close()
				return nil
			}
			return conn
		}

		log.Release("dial to %v error: %v", tcpConn.addr, err)
		time.Sleep(100 * time.Millisecond)
		continue
	}
}

func (tcpConn *TcpConn) GetAgent() Agent {
	return tcpConn.agent
}

func (tcpConn *TcpConn) Start() bool {
	if tcpConn.connType == CONN_TYPE_ACCEPT {
		return tcpConn.run()
	} else if tcpConn.connType == CONN_TYPE_CONNECT {
		return tcpConn.connect()
	} else {
		log.Error("tpc conn start err:%v connType:%v", "conn type wrong", tcpConn.connType)
		return false
	}
}

func (tcpConn *TcpConn) connect() bool {
	if tcpConn.agent == nil {
		log.Error("tcpConn run err:", "agent is nil")
		return false
	}
	//tcpConn.status = CONN_STATUS_CONNECT
	if !atomic.CompareAndSwapInt32(&tcpConn.cflag, 0, 1) {
		log.Error("tcpConn run err:", "cflag not 0")
		return false
	}
	tcpConn.cwg.Add(1)

	tool.Go(func() {
		defer func() {
			tcpConn.cwg.Add(-1)
		}()

		if tool.IsRuning() && tcpConn.cflag == 1 {
			conn := tcpConn.dial()
			if conn == nil {
				return
			}
			tcpConn.conn = conn
			tcpConn.run()

			//等待读写线程都退出
			tcpConn.rwg.Wait()
			tcpConn.conn.Close()
			tcpConn.conn = nil

			time.Sleep(CONNECT_INTERVAL)
		}
	})

	return true
}

func (tcpConn *TcpConn) run() bool {
	if tcpConn.conn == nil {
		log.Error("tpcConn run err:", "conn is nil")
		return false
	}
	if tcpConn.agent == nil {
		log.Error("tcpConn run err:", "agent is nil")
		return false
	}
	if !atomic.CompareAndSwapInt32(&tcpConn.rflag, 0, 1) {
		log.Error("tcpConn run err:", "rflag not 0")
		return false
	}
	tcpConn.rwg.Add(2)
	tool.Go(func() {
		defer func() {
			tcpConn.rwg.Add(-1)
			tcpConn.runOver() //让读写线程退出
		}()

		for tool.IsRuning() && tcpConn.rflag == 1 {
			data, err := tcpConn.ReadMsg()
			if err != nil {
				log.Error("tcpConn read err: %v", err)
				return
			}
			head, bytes, err := msg.Unmarshal(data)
			if err != nil {
				log.Error("tcpConn read err: %v", err)
			}
			if tcpConn.agent != nil {
				tcpConn.agent.Process(head, bytes)
			}
		}
	})

	tool.Go(func() {
		defer func() {
			tcpConn.rwg.Add(-1)
			tcpConn.runOver() //让读写线程退出
		}()

		for tool.IsRuning() && tcpConn.rflag == 1 {
			var data []byte
			select {
			case <-tool.GoSigChan():
			case data = <-tcpConn.writeChan:
				if data != nil {
					_, err := tcpConn.conn.Write(data)
					if err != nil {
						log.Error("tcpConn write err: %v", err)
						return
					}
				}
			}

			if data == nil {
				return
			}
		}
	})

	return true
}

func (tcpConn *TcpConn) runOver() {
	if atomic.CompareAndSwapInt32(&tcpConn.rflag, 1, 0) {
		tcpConn.conn.SetDeadline(time.Now())
		tcpConn.writeChan <- nil

		if tcpConn.server != nil {
			tcpConn.server.onCloseConn(tcpConn)
		}
		tcpConn.agent.OnClose()
	}
}

func (tcpConn *TcpConn) Close() {
	atomic.CompareAndSwapInt32(&tcpConn.cflag, 1, 0)
	if atomic.CompareAndSwapInt32(&tcpConn.rflag, 1, 0) {
		tcpConn.conn.SetDeadline(time.Now())
		tcpConn.writeChan <- nil

		if tcpConn.server != nil {
			tcpConn.server.onCloseConn(tcpConn)
			tcpConn.server = nil
		}
	}
}

func (tcpConn *TcpConn) Destroy() {
	tcpConn.rwg.Wait()
	tcpConn.cwg.Wait()
	if atomic.CompareAndSwapInt32(&tcpConn.dflag, 0, 1) {
		close(tcpConn.writeChan)
		if tcpConn.conn != nil {
			tcpConn.conn.Close()
			tcpConn.conn = nil
		}
	}
}

func (tcpConn *TcpConn) doWrite(b []byte) {
	if len(tcpConn.writeChan) == cap(tcpConn.writeChan) {
		log.Debug("close conn: channel full")
		tcpConn.Close()
		return
	}
	tcpConn.writeChan <- b
}

// b must not be modified by the others goroutines
func (tcpConn *TcpConn) Write(b []byte) {
	tcpConn.Lock()
	defer tcpConn.Unlock()
	if tcpConn.rflag != 1 || b == nil {
		log.Release("tcp conn write fail,rflag:%v b:%v", tcpConn.rflag, b)
		return
	}
	tcpConn.doWrite(b)
}

func (tcpConn *TcpConn) Read(b []byte) (int, error) {
	return tcpConn.conn.Read(b)
}

func (tcpConn *TcpConn) LocalAddr() net.Addr {
	return tcpConn.conn.LocalAddr()
}

func (tcpConn *TcpConn) RemoteAddr() net.Addr {
	return tcpConn.conn.RemoteAddr()
}

func (tcpConn *TcpConn) ReadMsg() ([]byte, error) {
	return tcpConn.msgParser.Read(tcpConn)
}

func (tcpConn *TcpConn) WriteMsg(args ...[]byte) error {
	return tcpConn.msgParser.Write(tcpConn, args...)
}
